package cn.codenest.util.impl;

import cn.codenest.util.RbmqService;
import com.rabbitmq.client.*;
import lombok.SneakyThrows;

import java.io.IOException;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.LockSupport;

/**
 * @author ：Hyman
 * @date ：Created in 2021/7/6 12:10
 * @description：
 * @modified By：
 * @version: $
 */
public class RbmqServiceImpl implements RbmqService {

    private String rabbitMqHost = "127.0.0.1";
    private String rabbitMqUsername = "guest";
    private String rabbitMqPassword = "guest";
    private Integer rabbitMqPort = 5672;
    private Integer rabbitMqTimeOut = 10000;


    @Override
    public void dealMsg() throws IOException, TimeoutException {
        // 创建连接和信道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(this.rabbitMqHost);
        factory.setPort(this.rabbitMqPort);
        factory.setConnectionTimeout(this.rabbitMqTimeOut);
        factory.setUsername(this.rabbitMqUsername);
        factory.setPassword(this.rabbitMqPassword);
        factory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            connection = factory.newConnection();
            channel = connection.createChannel();
            final Channel finalChannel = channel;
            channel.basicConsume("order.queue", false, new Consumer() {
                @Override
                public void handleConsumeOk(String s) {

                }

                @Override
                public void handleCancelOk(String s) {

                }

                @Override
                public void handleCancel(String s) throws IOException {

                }

                @Override
                public void handleShutdownSignal(String s, ShutdownSignalException e) {

                }

                @Override
                public void handleRecoverOk(String s) {

                }

                @Override
                public void handleDelivery(String s, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {
                    long deliveryTag = envelope.getDeliveryTag();
                    finalChannel.basicReject(deliveryTag, false);
                    System.out.println("拒绝接受消息：deliveryTag=" + deliveryTag);
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            LockSupport.park();
            if (channel != null && channel.isOpen()) {
                channel.close();
            }
            if (connection != null && connection.isOpen()) {
                connection.close();
            }
        }
    }

    @Override
    public void dealDlxMsg() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(this.rabbitMqHost);
        factory.setPort(this.rabbitMqPort);
        factory.setConnectionTimeout(this.rabbitMqTimeOut);
        factory.setUsername(this.rabbitMqUsername);
        factory.setPassword(this.rabbitMqPassword);
        factory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            connection = factory.newConnection();
            channel = connection.createChannel();
            final Channel finalChannel = channel;
            channel.basicConsume("dlx.queue", true, new Consumer() {
                @Override
                public void handleConsumeOk(String s) {

                }

                @Override
                public void handleCancelOk(String s) {

                }

                @Override
                public void handleCancel(String s) throws IOException {

                }

                @Override
                public void handleShutdownSignal(String s, ShutdownSignalException e) {

                }

                @Override
                public void handleRecoverOk(String s) {

                }

                @Override
                public void handleDelivery(String s, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {
                    String str = new String(bytes);
                    System.out.println("接受到死信消息：" + str);
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            LockSupport.park();
            if (channel != null && channel.isOpen()) {
                channel.close();
            }
            if (connection != null && connection.isOpen()) {
                connection.close();
            }
        }
    }

    @Override
    public void ttlMsg() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(this.rabbitMqHost);
        factory.setPort(this.rabbitMqPort);
        factory.setConnectionTimeout(this.rabbitMqTimeOut);
        factory.setUsername(this.rabbitMqUsername);
        factory.setPassword(this.rabbitMqPassword);
        factory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            connection = factory.newConnection();
            channel = connection.createChannel();
            final Channel finalChannel = channel;

            channel.queueDeclare("ttl.queue", true, false, false, null);
            channel.queueBind("ttl.queue", "order.exchange", "order.#");
            channel.basicConsume("ttl.queue", false, new Consumer() {
                @Override
                public void handleConsumeOk(String s) {

                }

                @Override
                public void handleCancelOk(String s) {

                }

                @Override
                public void handleCancel(String s) throws IOException {

                }

                @Override
                public void handleShutdownSignal(String s, ShutdownSignalException e) {

                }

                @Override
                public void handleRecoverOk(String s) {

                }

                @SneakyThrows
                @Override
                public void handleDelivery(String s, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {
                    /*在这里获取消息并且计算剩余的有效时间*/
                    long deliveryTag = envelope.getDeliveryTag();
                    finalChannel.basicReject(deliveryTag, true);
                    System.out.println("拒绝接受消息：deliveryTag=" + deliveryTag);
                    Thread.sleep(2000);
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            LockSupport.park();
            if (channel != null && channel.isOpen()) {
                channel.close();
            }
            if (connection != null && connection.isOpen()) {
                connection.close();
            }
        }
    }
}
